use chrono::{Days, NaiveDate};
use clickhouse_rs::{types::Value, Block};
use rust_decimal::Decimal;
use rust_decimal::prelude::ToPrimitive;

use crate::base::{get_ch_pool, get_my_pool, DiviType};
use sqlx::Row;
#[derive(Debug, Clone, PartialEq, PartialOrd)]
struct DiviInfo {
    dpid: u32,
    emid: u32,
    dsvl: f64,
    dsdt: NaiveDate,
    wodt: NaiveDate,
}
struct ZipperData {
    acid: u64,
    sdate: NaiveDate,
    edate: NaiveDate,
    dpids: Vec<u32>,
    emids: Vec<u32>,
    dsvls: Vec<f64>,
}
pub struct DiviZipper {
    divi_type: DiviType,
}
impl DiviZipper {
    pub (super) fn new(divi_type: DiviType) -> DiviZipper {
        DiviZipper { divi_type }
    }
    pub (super) async fn zip_divi(&self) -> anyhow::Result<Option<NaiveDate>> {
        if self.divi_type == DiviType::IndiClie|| self.divi_type == DiviType::CompBizFee|| self.divi_type == DiviType::IndiBizFee {
            return Ok(None);
        }
        let table_name = self.divi_type.divi_table_name();
        let sql = format!(
            r#"SELECT A.ACID,U.OrganID DPID,A.EMID,A.DSVL,A.DSDT DSDT,IFNULL(DATE_ADD(CAST(WODT AS DATE),INTERVAL -1 DAY),cast('2099-12-31' as DATE)) WODT,B.DSDT MIDT
	FROM {table_name} A INNER JOIN
	(SELECT ACID,MIN(CASE WHEN WODT IS NULL THEN DSDT ELSE WODT END) DSDT
	 FROM {table_name}
	 WHERE TMST>(SELECT TMST FROM CONF_UPDT) GROUP BY ACID) B ON A.ACID=B.ACID AND (A.WODT IS NULL OR A.WODT>=B.DSDT) INNER JOIN T_SysUser U ON A.EMID=U.ID
	WHERE A.ISWO=0 AND (A.DSDT<A.WODT OR A.WODT IS NULL)
	ORDER BY A.ACID,A.DSDT,A.DSVL DESC"#
        );
        let rows = sqlx::query(&sql).fetch_all(get_my_pool()).await?;
        let mut zip_datas: Vec<ZipperData> = Vec::with_capacity(10000);
        let mut acid: i32 = 0;
        let mut prev_acid: i32 = 0;
        let mut dis: Vec<DiviInfo> = Vec::new();
        let mut min_dsdt:Option<NaiveDate> = None;
        for row in rows.into_iter() {
            acid = row.try_get(0)?;
            let dpid: i32 = row.try_get(1)?;
            let emid: i32 = row.try_get(2)?;
            let dsvl: Decimal = row.try_get(3)?;
            let dsdt: NaiveDate = row.try_get(4)?;
            let wodt: NaiveDate = row.try_get(5)?;
            let midt: NaiveDate = row.try_get(6)?;
            let di = DiviInfo {
                dpid: dpid as u32,
                emid: emid as u32,
                dsvl: dsvl.to_f64().unwrap(),
                dsdt,
                wodt,
            };
            if acid != prev_acid {
                if prev_acid != 0 {
                    let mut dis2: Vec<DiviInfo> = Vec::new();
                    (dis2, dis) = (dis, dis2);
                    let res = self.zip_one(prev_acid as u64, dis2)?;
                    zip_datas.extend(res);
                }
                prev_acid = acid;
            }
            dis.push(di);
            if let Some(md)=min_dsdt {
                if md>midt{
                    min_dsdt=Some(midt);
                }
            }else{
                min_dsdt=Some(midt);
            }
        }
        if dis.len() > 0 {
            let res = self.zip_one(acid as u64, dis)?;
            zip_datas.extend(res);
        }
        self.create_tmp_table().await?;
        self.update_zip(zip_datas).await?;
        Ok(min_dsdt)
    }
    async fn create_tmp_table(&self) -> anyhow::Result<()> {
        let divi_name=&self.divi_type.divi_table_name()[2..];
        let tname = self.divi_type.join_table_name();
        if tname.is_empty(){
            return Ok(());
        }
        let sql = format!("CREATE TABLE IF NOT EXISTS {tname}(ACID UInt64,CUID UInt64,DSDT Date)ENGINE =  Join(ANY, LEFT, ACID)");
        let mut client=get_ch_pool().get_handle().await?;
        client.execute(&sql).await?;
        let sql=format!("truncate table {tname}");
        client.execute(sql).await?;
        let tname = format!("TMP_{divi_name}_ZIPPER");
        let sql = format!(r#"CREATE TABLE IF NOT EXISTS {tname}
        (
            SDATE Date COMMENT '开始日期',
            EDATE Date COMMENT '截止日期',
            ACID UInt64 COMMENT '账户ID',
            DIVI Nested
            (
                EMID UInt32,
                DPID UInt32,
                DSVL Float64
            )
        )
        ENGINE = MergeTree
        ORDER BY (SDATE,
        ACID)
        SETTINGS index_granularity = 8192"#);
        client.execute(&sql).await?;
        let sql=format!("truncate table {tname}");
        client.execute(sql).await?;
        Ok(())
    }
    async fn update_zip(&self, datas: Vec<ZipperData>) -> anyhow::Result<()> {
        if datas.len() == 0 {
            return Ok(());
        }
        let divi_name=&self.divi_type.divi_table_name()[2..];
        let zip_table_name = format!("TMP_{divi_name}_ZIPPER");
        let mut block = Block::new();
        for data in datas.into_iter() {
            let mut row:Vec<(String, Value)> = Vec::new();
            row.push(("SDATE".to_owned(),Value::from(data.sdate)));
            row.push(("EDATE".to_owned(),Value::from(data.edate)));
            row.push(("ACID".to_owned(),Value::from(data.acid)));
            row.push(("DIVI.EMID".to_owned(),Value::from(data.dpids)));
            row.push(("DIVI.DPID".to_owned(),Value::from(data.emids)));
            row.push(("DIVI.DSVL".to_owned(),Value::from(data.dsvls)));
            block.push(row)?;
        }
        let mut client=get_ch_pool().get_handle().await?;
        client.insert(&zip_table_name, block).await?;
        let acco_table_name = self.divi_type.acco_table_name();
        let back_table_name = self.divi_type.back_table_name();
        let sql = format!(r#"insert into {back_table_name}(ACID,CUID,DSDT)
	    SELECT A.ID ACID,A.CUID,B.SDATE FROM {acco_table_name} A INNER JOIN (SELECT ACID,MIN(SDATE) SDATE FROM {zip_table_name} GROUP BY ACID) B ON A.ID=B.ACID"#);
        client.execute(&sql).await?;
        let sql = format!("select count() from {back_table_name} where CUID=0");
        let b = client.query(&sql).fetch_all().await?;
        let count:u64=b.rows().next().unwrap().get(0)?;
        if count == 0 {
            log::warn!("发现表{back_table_name}存在CUID=0的情况，这会导致认定异常，请工程师处理");
        }
        let sql = format!(r#"insert into {zip_table_name}(SDATE,EDATE,ACID,DIVI.EMID,DIVI.DPID,DIVI.DSVL)
        SELECT SDATE,EDATE,ACID,DIVI.EMID,DIVI.DPID,DIVI.DSVL FROM T_{divi_name}_ZIPPER
        WHERE ACID NOT IN (SELECT ACID FROM {back_table_name})"#);
        client.execute(&sql).await?;
        let sql = format!(r#"insert into {zip_table_name}(SDATE,EDATE,ACID,DIVI.EMID,DIVI.DPID,DIVI.DSVL)
        WITH joinGet('{back_table_name}','DSDT',ACID) AS DSDT
        SELECT SDATE,if(EDATE>DSDT,addDays(DSDT,-1),EDATE) EDATE,ACID,DIVI.EMID,DIVI.DPID,DIVI.DSVL FROM T_{divi_name}_ZIPPER
        WHERE SDATE<DSDT"#);
        client.execute(&sql).await?;
        let sql = format!("DROP TABLE T_{divi_name}_ZIPPER");
        client.execute(&sql).await?;
        let sql = format!("RENAME TABLE TMP_{divi_name}_ZIPPER TO T_{divi_name}_ZIPPER");
        client.execute(&sql).await?;
        Ok(())
    }
    fn zip_one(&self, acid: u64, dis: Vec<DiviInfo>) -> anyhow::Result<Vec<ZipperData>> {
        let mut res: Vec<ZipperData> = Vec::new();
        let min_date = NaiveDate::from_ymd_opt(1970, 1, 1).unwrap();
        let mut prev_dsdt = &min_date;
        let mut dsdt_arr: Vec<&NaiveDate> = Vec::new();
        for di in dis.iter() {
            if di.dsdt > *prev_dsdt {
                dsdt_arr.push(&di.dsdt);
                prev_dsdt = &di.dsdt;
            }
        }
        let dlen = dsdt_arr.len();
        for i in 0..dlen {
            let sdate = dsdt_arr[i];
            let edate = if i + 1 == dlen {
                NaiveDate::from_ymd_opt(2099, 12, 31).unwrap()
            } else {
                dsdt_arr[i + 1].checked_sub_days(Days::new(1)).unwrap()
            };
            let mut zdata = ZipperData {
                acid,
                sdate: sdate.clone(),
                edate,
                dpids: Vec::new(),
                emids: Vec::new(),
                dsvls: Vec::new(),
            };
            for di in dis.iter() {
                if di.dsdt <= *sdate && di.wodt >= edate {
                    zdata.dpids.push(di.dpid);
                    zdata.emids.push(di.emid);
                    zdata.dsvls.push(di.dsvl);
                }
            }
            res.push(zdata);
        }
        Ok(res)
    }
}
